package com.wedding.mq;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import org.junit.Test;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

public class MQExchangeStudy {

	public static void main(String[] args) throws IOException, TimeoutException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		String exchange = "amq.direct";
		channel.exchangeDeclare(exchange, "direct", true);
		//4.声明
		//5.发送
		String msg = "test exchange of amq.direct";
		channel.basicPublish(exchange, "key1", null, msg.getBytes());
		channel.basicPublish(exchange, "key1", null, msg.getBytes());
//		channel.close();
//		connection.close();
		
	}
	@Test
	public void topicProduc() throws IOException, TimeoutException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		//4.声明
		String exchangeName = "ex_top";
		String routingKey1 = "usa.save";
		String routingKey2 = "usa.update";
		String routingKey3 = "uro.weather";
		//5.发送
		String msg = "Hello World RabbitMQ4 Direct Exchange Message";
		channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
		channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());
		channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());
	}
	@Test
	public void topicConsum() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		Map<String,Object> headers = new HashMap<>();
		headers.put("my1", "111");
		headers.put("my2", "222");
		//properties创建方法
		AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
				.deliveryMode(2)
				.contentEncoding("utf-8")
				.expiration("10000")
				.headers(headers)
				.build();
		
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		String exchangeName = "ex_top";
//		String exchangeType = "topic";
		String queueName = "q4test_usa";
		String routingKey = "usa.#";
		//表示声明了一个交换机
//		channel.exchangeDeclare(exchangeName, exchangeType,true,false,false,null);
		//表示声明了一个队列
		channel.queueDeclare(queueName,true,false,false,null);
		//建立一个绑定关系
		channel.queueBind(queueName, exchangeName, routingKey);
		//durable 是否持久化消息
		QueueingConsumer consumer = new QueueingConsumer(channel);
		//参数：队列名称,是否自动ACK,Consumer
		channel.basicConsume(queueName, true, consumer);
		while(true){
			//获取消息,如果没有消息,这一步将会一直阻塞
			Delivery delivery = consumer.nextDelivery();
			String msg = new String(delivery.getBody());
			System.out.println("收到消息:" + msg);
		}
	}
	@Test
	public void defaultExchange_producer() throws IOException, TimeoutException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		String queueName = "q4test_weather";
//        channel.queueDeclare(queueName, true, false, false, null);
        for (int i = 1; i <10; i++){  
            String message = "msg: try no exchange ";
            channel.basicPublish("", queueName ,null, message.getBytes());  
        }   
	}
	@Test
	public void defaultEchange_consumer() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		String queueName = "q4test_weather";
        //声明队列  
        channel.queueDeclare(queueName, true, false, false, null);  
        QueueingConsumer consumer = new QueueingConsumer(channel);  
        // 指定消费队列  
        String basicConsume = channel.basicConsume(queueName, true, consumer);  
        System.out.println(basicConsume);
        while (true){  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(message);
        }   
	}
	@Test
	public void fanoutProduc() throws IOException, TimeoutException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		//4.声明
		String exchangeName = "amq.fanout";
//		String queue = "q4test_fanout";
		channel.exchangeDeclare(exchangeName, "fanout", true);
//		channel.queueBind(queue, exchangeName, "abc");
		//5.发送
		String msg = "Hello World RabbitMQ4 study fanout";
		String routingKey1 = "33	";
		channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
//		channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
		channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
	}
	@Test
	public void fanoutConsumer() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		String queue = "q_noex";
		channel.queueDeclare(queue, true, false, false, null);
		QueueingConsumer consumer = new QueueingConsumer(channel);  
		// 指定消费队列  
		String basicConsume = channel.basicConsume(queue, true, consumer);
		System.out.println(basicConsume);
		while (true){  
		    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
		    String message = new String(delivery.getBody());
		    System.out.println(message);
		} 
		
		
	}
	@Test
	public void topicProduc2() throws IOException, TimeoutException {
		Connection connection = MQUtil.getConnection();
		Channel channel = connection.createChannel();
		//4.声明
		String exchangeName = "amq.topic";
		channel.exchangeDeclare(exchangeName, "topic", true);
		//5.发送
		String msg = "Hello World RabbitMQ4 study fanout";
		String routingKey1 = "usa.threefour.k";
		channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
	}
	
	
}
